//Linux C plus plus file
/************************************************************************
*
*
*       			FileName:listenThread.cpp
*
*         			Author.wzj
*
*       			Email:345517039@qq.com
*
*      	    		Create:2017-06-22 23:34:32
*
*      		    	Description:-
*
*
*************************************************************************/
#include "listenThread.h"



/**
 * @brief 构造函数初始化;
 */
ListenThread::ListenThread(){
	memset(&serconf, 0, sizeof(struct server_conf_st));
}


/**
 * @brief:
 * @para:
 * @author:
 * @return value:
 */
ListenThread::~ListenThread(){
	//资源的释放;
	
}


/**
 * @brief 监听线程的初始化接口 
 * @para: void *arg: 参数信息
 * @author: wzj 20171206
 * @return value: RET::FAIL 失败, RET::SUCCESS 成功
 */
int ListenThread::init(void *arg){
#define LOG_BUF_MAX 1024
	std::vector<int> *local = (std::vector<int > *)arg;
	int ret = RET::FAIL;
	ret = confMgr::getConfValue(SYSTEM_CONFIG, &serconf, conf);	
	if ( ret == RET::FAIL){
		fprintf(stderr, "[ ERROR ]LISTEN THREAD INIT GET CONFIG OPTION FAILURE!\n");
		return ret;
	}
	lis.uListenSocketCount = serconf.listenThreadNum;
	this->mem.init(THREAD_LEVEL);
	lis.io = new epollModel();
	lis.ioType=EPOLL_MODEL;
	ret = lis.io->init();
	if (ret != RET::SUCCESS){
		fprintf(stderr, "[ ERROR ]LISTEN THREAD INIT IO MODEL FAILURE!\n");
		return RET::FAIL;
	}
	for(size_t i = 0; i < lis.uListenSocketCount; i++){
		struct socket_st * st = (struct socket_st * )mem.allocateMemFromMempool(sizeof(struct socket_st));
		if ( !st ){

			return RET::FAIL;
		}
		//初始化;
		st->fd=socket(AF_INET,SOCK_STREAM,0);
		st->addr.sin_family=AF_INET;
		st->addr.sin_port=htons(8080);
		st->addr.sin_addr.s_addr=inet_addr("172.17.0.1");
		if(this->lis.head == NULL){ 
			this->lis.head = st;
		}else{
			st->next = this->lis.head;
			this->lis.head = st;
		}
		//设置成端口可以重用;
		int optval = 1;
		socklen_t len = sizeof(optval);
		ret = setsockopt(st->fd, SOL_SOCKET, SO_REUSEADDR, &optval, len);
		if ( ret == RET::FAIL ){
			fprintf(stderr,"[ ERROR ]LISTEN THREAD SET SOCKET REUSEADDR ERROR!\n");
			return ret;
		}
		//进行bind绑定;
		ret = bind(st->fd, (struct sockaddr *)&(st->addr), sizeof(st->addr));
		if (ret == RET::FAIL ){
			fprintf(stderr,"[ ERROR ]LISTEN THREAD BIND SOCKET ERROR!\n");
			return ret;
		}
		lis.io->addEvent(st,LISTEN_SOCKET);
		ret = listen(st->fd, 5);
	}
	lis.currentIndex = 0;
	//添加标准输入;
	lis.io->addEvent(fileno(stdin));
	//加入本地套接字;
	for(size_t index = 0; index < local->size(); index++){
		if ( local->at(index)!= -1){
			if (lis.logfd == 0 ){
				lis.logfd = local->at(index);
				DEBUG_TRACE(2,("[ INFO ]LISTEN THREAD LOG FD:%d\n", lis.logfd));
			}else{
				lis.socketqueue.push_back(local->at(index));
				DEBUG_TRACE(2,("[ INFO ]LISTEN THREAD SOCKET TO WORKER THREAD:%u\n", local->at(index)));
			}
			lis.io->addEvent(local->at(index));//最好设置成为非阻塞的;
		}
	}
	lis.logbuf = (char *)mem.allocateMemFromMempool(LOG_BUF_MAX);
	if( !lis.logbuf ){
		fprintf(stderr, "[ ERROR ]ALLOC MEM FROM MEMPOOL OCCUR ERROR!\n");
		return RET::FAIL;
	}
	lis.bufSize = LOG_BUF_MAX;
	lis.bStopFlag = false;
	//线程参数初始化;	
	lis.pth.func = ListenThread::listenThreadRun;
	lis.pth.arg = (void *)this;
	ret = pthread_create(&(lis.pth.id),&(lis.pth.attr),lis.pth.func,lis.pth.arg);
	if ( ret != 0 ){
		fprintf(stderr, "[ ERROR ]LISTEN THREAD CREATE FAILURE!\n");
		//写入错误日志;
		return RET::FAIL;
	}
	LogManger::LogFormatAndSend(lis.logfd, INIT , lis.logbuf, "sus", "LISTEN THREAD:", (unsigned int)lis.pth.id, "CREATE SUCCESSFULLY!");
	DEBUG_TRACE(2,("[ INFO ]LISTEN THREAD TID:%lu CREATE SUCCESSFULLY!\n",lis.pth.id));
	return ret;
}

/**
 * @brief 
 *
 * @param count
 *
 * @return 
 */
int ListenThread::handleEventForIo(int count){
#define CMD_BUF_SIZE 100
	char cmdLine[CMD_BUF_SIZE];
	memset(cmdLine, 0, CMD_BUF_SIZE);
	char *cmd = NULL;
	int iRet = RET::FAIL;
	switch(lis.ioType)
	{
		case EPOLL_MODEL:
			{
				EPOLL_CTX *ctx = (EPOLL_CTX *)(lis.io->getIoCtx());
				for ( int i = 0; i < count; i++ ){
					if( ctx->events[i].data.fd  == 0 ){  // 标准输入;
						cmd = fgets(cmdLine, CMD_BUF_SIZE, stdin);
						if (!cmd){
							return iRet;
						}else{
							std::vector<char *> veccmd;
							conf->getConfMgr().parseCmd(cmdLine,veccmd);
							fprintf(stdout,"[SYSTEM]");
							fflush(stdout);
							break;
						}
					}
					if ( ctx->events[i].data.fd == lis.head->fd ){  // listen socket;
						//accept a connection from client and insert this
						//connection  into queue;
						struct sockaddr_in addr;
						socklen_t len = sizeof(addr);
						int connectfd = accept(lis.head->fd, (struct sockaddr *)&addr, &len);
						if ( connectfd < 0 ){
							fprintf(stderr, "[ ERROR ] LISTEN THREAD ACCEPT A BAD CONNECTION IN FUNCTION:%s LINE:%d\n", __func__, __LINE__);
							return RET::FAIL;
						}
						//random a socketpair index and insert a connection into queue;
						if ( lis.currentIndex == lis.socketqueue.size()){
							lis.currentIndex = 0;
						}
						iRet = write(lis.socketqueue[lis.currentIndex++], &connectfd, sizeof(connectfd));
						if ( iRet <= 0 ){
							return RET::FAIL;
						}
						LogManger::LogFormatAndSend(lis.logfd, NETWORKCONNECTION, lis.logbuf, "sds", "LISTEN THREAD SEND A CONNECTION:", connectfd, "TO WORKER THREAD");
						DEBUG_TRACE(2,("[ CONNECTION ] LISTEN THREAD GET A CONNECTION:%d FROM CLIENT!\n", connectfd));
					}

				}
			}
			break;
		case SELECT_MODEL:
			{

			}
			break;
		case POLL_MODEL:
			{

			}
			break;
		default:
			{
				fprintf(stderr,"[ ERROR ]UNKNOWN IO MODEL TYPE!\n");
				return RET::FAIL;
			}
	}
	return RET::SUCCESS;
}




/**
 * @brief 监听线程的线程回调函数;
 *
 * @param arg
 *
 * @return 
 */
void *ListenThread::listenThreadRun(void *arg){
	prctl(PR_SET_NAME,"listen_thread");	
	int ioCount = 0;
	int iRet = RET::FAIL;
	ListenThread * lis = (ListenThread *)arg;
	assert(lis!=NULL);
	struct listenThread_ctx ctx = lis->getListenThreadOpt();	
	//这边就开始搞一下;	
	fprintf(stdout,"[SYSTEM]");
	fflush(stdout);
	while(!ctx.bStopFlag){
		//超时事件;	
		//读写事件;
		ioCount = ctx.io->waitForEvent();
		if (ioCount == RET::FAIL){
			continue;
		}else if(ioCount == 0){
			continue;
		}else{
			iRet = lis->handleEventForIo(ioCount);
			if(iRet == RET::FAIL){
				DEBUG_TRACE(2,("[ ERROR ]LISTEN THREAD EXIT ABNORMALLY!\n"));
				ctx.bStopFlag=1;
			}
		}
	}
	return lis;
}

